const {
	OnExtensionSubscriber,
	OnNextSubscriber,
	OnTerminalSubscriber,
	Payload,
	RSocket,
	RSocketConnector,
	RSocketServer,
} = require("./@rsocket/core");
const {
	TcpClientTransport
} = require("./@rsocket/transport-tcp-client");
const {
	WellKnownMimeType,
	encodeRoute,
	encodeCompositeMetadata
} = require("./@rsocket/composite-metadata");
const MESSAGE_RSOCKET_ROUTING = WellKnownMimeType.MESSAGE_RSOCKET_ROUTING;
const Logger = require("../setLog.js");
class ReconnectableRSocket {
	options;
	rsocket;
	activeType;
	constructor(options) {
		this.options = options;
		this.activeType = -1;
		// this.connect();
	}
	makeConnector(type) {
		let _this = this;
		let mimeTypeList = ['application/json', 'application/octet-stream'];
		return new RSocketConnector({
			setup: {
				dataMimeType: mimeTypeList[type], //application/octet-stream
				keepAlive: 1000000, // avoid sending during test
				lifetime: 100000,
				// metadataMimeType: 'text/plain',
				metadataMimeType: 'message/x.rsocket.routing.v0',
			},
			transport: new TcpClientTransport({
				connectionOptions: {
					host: _this.options.host,
					port: _this.options.port,
				},
			}),
		});
	}
	async connect(type) {
		let _this = this;
		_this.activeType = type;
		const connector = this.makeConnector(type);
		this.rsocket = await connector.connect();
		// connector.connect().then(socket=>{
		// 	console.log("连接成功")
		// 	_this.rsocket=socket;
		// },error =>{
		// 	console.log("连接出错")
		// 	console.log(error)
		// })
	}
	closeSocket() {
		let _this = this;
		if (_this.rsocket) {
			_this.rsocket.close();
			_this.rsocket = null;
		}
	}
	getMatadata(route,fileName) {
		let compositeMetaData = undefined;
		if(fileName){
			let name=Buffer.from(fileName).toString('base64');
			route+="."+name;
		}
		return Buffer.from(String.fromCharCode(route.length)+route);
	}
	async requestResponse(data, rote, fileName, type) {
		let _this = this;
		if (!type) {
			type = 0;
		}
		if (!_this.rsocket) {
			Logger.info("连接关闭，重新连接")
			await _this.connect(type);
			// throw new Error('Not Connected yet. Retry later');
		} else if (_this.activeType != type) {
			_this.closeSocket();
			await _this.connect(type);
		}
		let buffData;
		if (type) {
			buffData = data;
		} else {
			buffData = Buffer.from(data);
		}
		let matadata = _this.getMatadata(rote, fileName);
		return new Promise((resolve, reject) => {
			_this.rsocket.requestResponse({
				data: buffData,
				// metadata:matadata,
				metadata: matadata,
			}, {
				onError: (e) => {
					Logger.error("发送出错")
					Logger.error(e)
					reject(e);
				},
				onNext: (payload, isComplete) => {
					Logger.info("next")
					Logger.info(
						`payload[data: ${payload.data}; metadata: ${payload.metadata}]|${isComplete}`
					);
					resolve(payload.data.toString());
				},
				onComplete: () => {
					Logger.info("onComplete")
				},
				onExtension: () => {
					Logger.info("onExtension")
				},
			});
		});
	}
	async requestStream(data, rote, fileName) {
		let _this = this;
		if (!_this.rsocket) {
			Logger.info("连接关闭，重新连接")
			await _this.connect(1);
			// throw new Error('Not Connected yet. Retry later');
		} else if (_this.activeType != 1) {
			_this.closeSocket();
			await _this.connect(1);
		}
		let length = data.length;
		let burrers = new Array();
		const unit_buffer = 40;
		const maxPayloads = Math.ceil(length / unit_buffer);
		// for(let i=0;i<maxPayloads;i++){
		// 	if(i==maxPayloads-1){
		// 		burrers.push(data.slice(i*unit_buffer,length));
		// 	}else{
		// 		burrers.push(data.slice(i*unit_buffer,(i+1)*unit_buffer))
		// 	}
		// }
		// console.log(burrers);
		let matadata = _this.getMatadata(rote, fileName);
		return new Promise((resolve, reject) => {
			let payloadsReceived = 0;
			const maxPayloads = 6;
			const requester = _this.rsocket.requestStream({
					data: data,
					metadata: matadata,
				},
				3, {
					onError: (e) => {
						console.log("error~~~~~~~~~~~~")
						console.log(e);
						reject(e)
					},
					onNext: (payload, isComplete) => {
						Logger.info(
							`[client] payload[data: ${payload.data}; metadata: ${payload.metadata}]|isComplete: ${isComplete}`
						);

						payloadsReceived++;
						console(payloadsReceived);
						// request 5 more payloads event 5th payload, until a max total payloads received
						if (payloadsReceived % 2 == 0 && payloadsReceived < maxPayloads) {
							requester.request(2);
						} else if (payloadsReceived > maxPayloads) {
							requester.cancel();
							setTimeout(() => {
								resolve(null);
							});
						}
						if (isComplete) {
							console("完成");
							resolve(null);
						}
					},
					onComplete: () => {
						resolve(null);
					},
					onExtension: () => {},
				}
			);
		})
	}
	async requestChannel(data, rote, fileName) {
		let _this = this;
		if (!_this.rsocket) {
			Logger.info("连接关闭，重新连接")
			await _this.connect(1);
			// throw new Error('Not Connected yet. Retry later');
		} else if (_this.activeType != 1) {
			_this.closeSocket();
			await _this.connect(1);
		}
		let matadata = _this.getMatadata(rote,  fileName);
		return new Promise((resolve, reject) => {
			const requester = _this.rsocket.requestChannel({
					data: data,
					metadata: matadata,
				},
				1,
				false, {
					onError: (e) => reject(e),
					onNext: (payload, isComplete) => {
						console.log(
							`payload[data: ${payload.data}; metadata: ${payload.metadata}]|${isComplete}`
						);
						requester.request(1);
						if (isComplete) {
							resolve(payload);
						}
					},
					onComplete: () => {
						resolve(null);
					},
					onExtension: () => {},
					request: (n) => {
						console.log(`request(${n})`);
						requester.onNext({
								data: Buffer.from("Message"),
							},
							true
						);
					},
					cancel: () => {},
				}
			);
		});
	}

}

async function main() {
	const rsocket = new ReconnectableRSocket({
		host: "192.168.11.8",
		port: "8888"
	})
	let param = {
		url: "login",
		data: {
			usename: "admin",
			password: "admin"
		}
	}
	let res = await rsocket.requestStream(JSON.stringify(param));
	Logger.info("res")
	Logger.info(res);

}
// main()
module.exports = ReconnectableRSocket;
